package edu.buaa.traffic.client;

import com.google.common.util.concurrent.ListenableFuture;
import edu.buaa.common.client.MariaDbExecutorClient;
import edu.buaa.common.transaction.AbstractTransaction;
import edu.buaa.traffic.model.StatusUpdate;
import edu.buaa.traffic.transaction.*;
import edu.buaa.traffic.transaction.internal.NodeNeighborRoadTx;
import edu.buaa.traffic.algo.GeneralizedDijkstra;
import edu.buaa.traffic.algo.Topology;
import edu.buaa.utils.Pair;
import edu.buaa.utils.TimeTicker;
import edu.buaa.utils.Triple;
import scala.Tuple3;
import scala.Tuple6;

import java.sql.*;
import java.util.*;
import java.util.concurrent.*;

/**
 * Created by crusher. 2020.10.14
 */
public class MariaDBExecutorClient extends MariaDbExecutorClient {
    private final HashMap<String, Long> name2Id = new HashMap<>();
    private final HashMap<Long, String> id2Name = new HashMap<>();
    // pair(id, (r_id, st_time, en_time, status, travel_t, seg_cnt))
    private final ArrayList<Tuple6<Long, Long, Long, Integer, Integer, Integer>> buffer = new ArrayList<>();
    // pair(rid, (r_id, st_time, en_time, status, travel_t, seg_cnt))
    private final HashMap<Long, Tuple6<Long, Long, Long, Integer, Integer, Integer>> lastRIdBuffer = new HashMap<>();
    private final TimeTicker tick = new TimeTicker(30, 200);

    public MariaDBExecutorClient(String serverHost, int parallelCnt) throws Exception {
        super(serverHost, "traffic", parallelCnt);
    }

    @Override
    protected List<String> createTables() {
        return Arrays.asList(
                "CREATE table cross_node(cross_id bigint PRIMARY KEY, name varchar(255))",
                "CREATE table road(r_id bigint primary key, r_address CHAR(12), r_start bigint, r_end bigint, r_length INT, r_type INT)",
                "CREATE table temporal_status(ts_id bigint PRIMARY KEY AUTO_INCREMENT, st_time TIMESTAMP(0), en_time TIMESTAMP(0), r_id bigint, status INT, travel_t INT, seg_cnt INT, PERIOD FOR time_period(st_time, en_time))",
                "create index ind_rid_st_en on temporal_status(r_id, st_time, en_time)"
                );
    }

    @Override
    protected void connected(Connection conn) throws Exception {
        buildMapBetweenIdAndName(conn);
        System.out.println("Build the map!");
    }

    @Override
    public ListenableFuture<ServerResponse> execute(AbstractTransaction tx) throws InterruptedException {
        switch (tx.getTxType()) {
            case tx_import_static_data:
                return this.submit(execute((ImportStaticDataTx) tx));
            case tx_import_temporal_data:
                return this.submit(execute((ImportTemporalDataTx) tx), tx.getSection());
            case tx_query_road_by_temporal_condition:
                return this.submit(execute((EntityTemporalConditionTx) tx));
            case tx_query_node_neighbor_road:
                return this.submit(execute((NodeNeighborRoadTx) tx));
            case tx_query_snapshot:
                return this.submit(execute((SnapshotQueryTx) tx));
            case tx_update_temporal_data:
                return this.submit(execute((UpdateTemporalDataTx) tx));
            case tx_query_snapshot_aggr_max:
                return this.submit(execute((SnapshotAggrMaxTx) tx));
            case tx_index_tgraph_aggr_duration:
                return this.submit(execute((SnapshotAggrDurationTx) tx));
            case tx_query_reachable_area:
                return this.submit(execute((ReachableAreaQueryTx) tx));
            default:
                throw new UnsupportedOperationException();
        }
    }

    private void buildMapBetweenIdAndName(Connection conn) throws SQLException {
        try (PreparedStatement stmt0 = conn.prepareStatement("select r_id, r_address from road")) {
            ResultSet rs = stmt0.executeQuery();
            while (rs.next()) {
                Long id = rs.getLong("r_id");
                String name = rs.getString("r_address");
                id2Name.put(id, name);
                name2Id.put(name, id);
            }
        }
    }

    private Req execute(ImportStaticDataTx tx) {
        return conn -> {
            try (PreparedStatement stat1 = conn.prepareStatement("INSERT INTO cross_node VALUES (?, ?)");
                 PreparedStatement stat2 = conn.prepareStatement("INSERT INTO road VALUES (?, ?, ?, ?, ?, ?)")) {
                conn.setAutoCommit(false);
                for (ImportStaticDataTx.StaticCrossNode p : tx.getCrosses()) {
                    stat1.setLong(1, p.getId());
                    stat1.setString(2, p.getName());
                    stat1.addBatch();
                }
                stat1.executeBatch();

                for (ImportStaticDataTx.StaticRoadRel r : tx.getRoads()) {
                    stat2.setLong(1, r.getRoadId());
                    stat2.setString(2, r.getId());
                    stat2.setLong(3, r.getStartCrossId());
                    stat2.setLong(4, r.getEndCrossId());
                    stat2.setInt(5, r.getLength());
                    stat2.setInt(6, r.getType());
                    stat2.addBatch();
                }
                stat2.executeBatch();
                conn.commit();
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                conn.setAutoCommit(true);
            }
            buildMapBetweenIdAndName(conn);
            return new AbstractTransaction.Result();
        };
    }

    public Req execute(ImportTemporalDataTx tx) {
        return conn -> {
            conn.setAutoCommit(false);
            try (Statement stmt0 = conn.createStatement()) {
                for(StatusUpdate s :tx.getData()){
                    StringBuilder sql = new StringBuilder();
                    sql.append("UPDATE temporal_status FOR PORTION OF time_period FROM '")
                            .append(new Timestamp(s.getTime() * 1000L)).append("' TO '")
                            .append(new Timestamp((Integer.MAX_VALUE-2) * 1000L)).append("' SET ")
                            .append("status=").append(s.getJamStatus())
                            .append(",travel_t=").append(s.getTravelTime())
                            .append(",seg_cnt=").append(s.getSegmentCount());
                    sql.append(" WHERE r_id=").append(name2Id.get(s.getRoadId()));
                    if(tick.shouldTick()) System.out.println(sql);
                    stmt0.addBatch(sql.toString());
                }
                stmt0.executeBatch();
            } catch (SQLException e) {
                e.printStackTrace();
            }
            return new AbstractTransaction.Result();
        };
    }

    private Req execute(UpdateTemporalDataTx tx) {
        return conn -> {
            conn.setAutoCommit(true);
            StringBuilder sql = new StringBuilder();
            sql.append("UPDATE temporal_status FOR PORTION OF time_period FROM '")
                    .append(new Timestamp(tx.getStartTime() * 1000L)).append("' TO '")
                    .append(new Timestamp((tx.getEndTime()+1) * 1000L)).append("' SET ")
                    .append("status=").append(tx.getJamStatus())
                    .append(",travel_t=").append(tx.getTravelTime())
                    .append(",seg_cnt=").append(tx.getSegmentCount());
            sql.append(" WHERE r_id=").append(name2Id.get(tx.getRoadId()));
            System.out.println(sql);
            try (PreparedStatement stmt0 = conn.prepareStatement(sql.toString())) {
                stmt0.executeQuery();
            } catch (SQLException e) {
                e.printStackTrace();
            }
            return new AbstractTransaction.Result();
        };
    }

    private ArrayList<Long> getRoadIdList(Connection conn) throws SQLException {
        ArrayList<Long> res = new ArrayList<>();
        try (PreparedStatement stat = conn.prepareStatement("select r_id from road where 1 = 1")) {
            ResultSet rs = stat.executeQuery();
            while (rs.next()) {
                res.add(rs.getLong("r_id"));
            }
        }
        return res;
    }


    private Req execute(SnapshotQueryTx tx) {
        return conn -> {
            List<Pair<String, Integer>> res = new ArrayList<>();
            List<Long> id = getRoadIdList(conn);
            String sql = "select " + tx.getPropertyName() + " as property from temporal_status where ts_id = (" +
                    "select ts_id from temporal_status where r_id = ? and \"" + new Timestamp((long) tx.getTimestamp() * 1000l) +
                    "\" between st_time and en_time)";
            System.out.println(sql);
            for (long rId : id) {
                try (PreparedStatement stat = conn.prepareStatement(sql)) {
                    stat.setLong(1, rId);
                    ResultSet rs = stat.executeQuery();
                    if (rs.next()) {
                        res.add(Pair.of(id2Name.get(rId), rs.getInt("property")));
                    }
                }
            }
            SnapshotQueryTx.Result result = new SnapshotQueryTx.Result();
            result.setRoadStatus(res);
            return result;
        };
    }


    private Req execute(SnapshotAggrMaxTx tx) {
        return conn -> {
            List<Pair<String, Integer>> res = new ArrayList<>();
            String sql = "select r_id, max(" + tx.getP() + ") as max_p from temporal_status where ts_id in (select ts_id from temporal_status where r_id = ? " +
                    "and st_time <= \"" + new Timestamp((long) tx.getT1() * 1000l) + "\" and en_time >= \"" + new Timestamp((long) tx.getT0() * 1000l) + "\") group by r_id";
            System.out.println(sql);
            ArrayList<Long> id = getRoadIdList(conn);
            for (long rId : id) {
                try (PreparedStatement stat = conn.prepareStatement(sql)) {
                    conn.setAutoCommit(false);
                    stat.setLong(1, rId);
                    ResultSet rs = stat.executeQuery();
                    if (rs.next()) {
                        res.add(Pair.of(id2Name.get(rId), rs.getInt("max_p")));
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            SnapshotAggrMaxTx.Result result = new SnapshotAggrMaxTx.Result();
            result.setRoadTravelTime(res);
            return result;
        };
    }

    private Req execute(SnapshotAggrDurationTx tx) {
        return conn -> {
            List<Triple<String, Integer, Integer>> res = new ArrayList<>();
            ArrayList<Long> id = getRoadIdList(conn);
            int t0 = tx.getT0(), t1 = tx.getT1();
            String sql = "select " + tx.getP() + " as property, st_time, en_time from temporal_status where ts_id in " +
                    "(select ts_id from temporal_status where r_id = ? and st_time <= \"" +
                    new Timestamp((long) t1 * 1000l) + "\" and en_time >= \"" +
                    new Timestamp((long) t0 * 1000l) + "\") order by property";
            System.out.println(sql);
            for (long rId : id) {
                try (PreparedStatement stat = conn.prepareStatement(sql)) {
                    stat.setLong(1, rId);
                    ResultSet rs = stat.executeQuery();
                    int lastValue = -1, total = 0;
                    String rName = id2Name.get(rId);
                    while (rs.next()) {
                        int value = rs.getInt("property");
                        int st = (int) (rs.getTimestamp("st_time").getTime() / 1000l);
                        int en = (int) (rs.getTimestamp("en_time").getTime() / 1000l);
                        int duration;
                        if (t0 >= st && t1 <= en) { // inner
                            duration = t1 - t0;
                        } else if (t0 >= st) { // left
                            duration = en - t0 + 1;
                        } else if (t1 >= en) { // middle
                            if (t1 == en) duration = t1 - st;
                            else duration = en - st + 1;
                        } else { // right
                            duration = t1 - st;
                        }
                        if (lastValue == -1 || value == lastValue) {
                            total += duration;
                        } else {
                            res.add(Triple.of(rName, lastValue, total));
                            total = duration;
                        }
                        lastValue = value;
                    }
                    if (lastValue != -1) res.add(Triple.of(rName, lastValue, total));
                }
            }
            SnapshotAggrDurationTx.Result result = new SnapshotAggrDurationTx.Result();
            result.setRoadStatDuration(res);
            return result;
        };
    }

    private Req execute(EntityTemporalConditionTx tx) {
        return conn -> {
            List<String> res = new ArrayList<>();
            ArrayList<Long> id = getRoadIdList(conn);
            String sql = "select ts_id from temporal_status where r_id = ? and st_time <= \"" + new Timestamp((long) tx.getT1() * 1000l) +
                    "\" and en_time >= \"" + new Timestamp((long) tx.getT0() * 1000l) + "\" and " + tx.getP() +
                    " between " + tx.getVmin() + " and " + tx.getVmax() + " limit 1";
            System.out.println(sql);
            for (long rId : id) {
                try (PreparedStatement stat = conn.prepareStatement(sql)) {
                    stat.setLong(1, rId);
                    ResultSet rs = stat.executeQuery();
                    if (rs.next()) {
                        res.add(id2Name.get(rId));
                    }
                }
            }

            EntityTemporalConditionTx.Result result = new EntityTemporalConditionTx.Result();
            result.setRoads(res);
            return result;
        };
    }

    private Req execute(NodeNeighborRoadTx tx) {
        return conn -> {
            List<String> res = new ArrayList<>();
            try (Statement stat = conn.createStatement()) {
                conn.setAutoCommit(true);
                String sql = "select r_name from road where r_start = " + tx.getNodeId() + " or r_end = " + tx.getNodeId();
                ResultSet rs = stat.executeQuery(sql);
                while (rs.next()) {
                    res.add(rs.getString("r_name"));
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
            NodeNeighborRoadTx.Result result = new NodeNeighborRoadTx.Result();
            result.setRoadIds(res);
            return result;
        };
    }


    private Req execute(ReachableAreaQueryTx tx) {
        return conn -> GeneralizedDijkstra.solve(tx, new MariaDBTopology(conn));
    }

    private static class MariaDBTopology implements Topology {

        private final Connection conn;

        MariaDBTopology(Connection conn) {
            this.conn = conn;
        }

        @Override
        public ArrayList<Pair<Long, Long>> getRoadList(long curCross) {
            ArrayList<Pair<Long, Long>> ret = new ArrayList<>();
            try (PreparedStatement stmt = conn.prepareStatement("select r_id, r_end from road where r_start = ?")) {
                stmt.setLong(1, curCross);
                ResultSet rs = stmt.executeQuery();
                while (rs.next()) ret.add(Pair.of(rs.getLong("r_id"), rs.getLong("r_end")));
            } catch (SQLException e) {
                e.printStackTrace();
            }
            return ret;
        }

        @Override
        public int earliestArriveTime(long roadId, int when, int until) {
            int ret = Integer.MAX_VALUE;
            try (PreparedStatement stmt = conn.prepareStatement("select st_time, en_time, travel_t from temporal_status where r_id = ? and st_time <= ? and en_time >= ?")) {
                stmt.setLong(1, roadId);
                stmt.setTimestamp(2, new Timestamp(until * 1000L));
                stmt.setTimestamp(3, new Timestamp(when * 1000L));
                ResultSet rs = stmt.executeQuery();
                ArrayList<Tuple3<Timestamp, Timestamp, Integer>> l = new ArrayList<>();
                while (rs.next())
                    l.add(Tuple3.apply(rs.getTimestamp("st_time"), rs.getTimestamp("en_time"), rs.getInt("travel_t")));
                int nextWhen = when;
                for (Tuple3<Timestamp, Timestamp, Integer> ele : l) {
                    int stTime = (int) (ele._1().getTime() / 1000L);
                    int enTime = (int) (ele._2().getTime() / 1000L);
                    int travelTime = ele._3();
                    if (when <= enTime && until <= enTime) // inner.
                        return when + travelTime <= until ? when + travelTime : Integer.MAX_VALUE;
                    if (stTime > when) nextWhen = stTime; // middle and right.
                    int cur = nextWhen + travelTime;
                    if (cur <= enTime && cur <= until && cur < ret) ret = cur;
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
            return ret;
        }
    }

}

